package sunyu.tools.hbase;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Component;
import sunyu.tools.hbase.pojo.Row;
import sunyu.tools.hbase.rowmapper.HbaseRowMapper;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import static org.apache.ibatis.io.Resources.getCharset;

/**
 * Hbase工具类
 *
 * @author 孙宇
 */
@Component
public class HbaseTools {

    public static final String COPROCESSCLASSNAME = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";// 协处理器
    public static final int TOTALUSEPAGESIZE = 500000;// 统计结果集总数的时候，使用的分页limit
    private static final Logger logger = LoggerFactory.getLogger(Thread.currentThread().getClass());
    private static final HbaseRowMapper hbaseRowMapper = new HbaseRowMapper();
    @Resource(name = "hbaseTemplate")
    private HbaseTemplate template;

    /**
     * 创建表
     *
     * @param tableName 表名
     * @param family    列族
     *
     * @return 是否创建成功
     */
    public Boolean createTable(String tableName,
                               String family) {
        boolean b = false;
        Configuration conf = template.getConfiguration();
        Connection connection = null;
        Admin admin = null;
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            if (!admin.tableExists(TableName.valueOf(tableName))) {
                HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
                desc.addFamily(new HColumnDescriptor(family));
                admin.createTable(desc);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                admin.close();
                b = true;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return b;
    }

    /**
     * 删除表
     *
     * @param tableName
     *
     * @throws IOException
     */
    public void deleteTable(String tableName) {
        Configuration conf = template.getConfiguration();
        Connection connection = null;
        Admin admin = null;
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                admin.disableTable(TableName.valueOf(tableName));// 禁用后删除
                admin.deleteTable(TableName.valueOf(tableName));// 删除
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 激活表
     *
     * @param tableName
     *
     * @throws IOException
     */
    public void enableTable(String tableName) {
        Configuration conf = template.getConfiguration();
        Connection connection = null;
        Admin admin = null;
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                admin.enableTable(TableName.valueOf(tableName));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 禁用表
     *
     * @param tableName
     *
     * @throws IOException
     */
    public void disableTable(String tableName) {
        Configuration conf = template.getConfiguration();
        Connection connection = null;
        Admin admin = null;
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                admin.disableTable(TableName.valueOf(tableName));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 对table注册协处理器Coprocessor
     *
     * @param tableName
     *
     * @throws IOException
     */
    public void registerCoprocessor(String tableName) {
        Configuration conf = template.getConfiguration();
        Connection connection = null;
        Admin admin = null;
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                admin.disableTable(TableName.valueOf(tableName));// 禁用后才能修改
                try {
                    HTableDescriptor htd = admin.getTableDescriptor(TableName.valueOf(tableName));
                    htd.addCoprocessor(COPROCESSCLASSNAME);
                    admin.modifyTable(TableName.valueOf(tableName), htd);// 修改
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    admin.enableTable(TableName.valueOf(tableName));// 解除禁用
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 插入一行
     *
     * @param tableName  表名
     * @param rowName    rowKey
     * @param familyName 列族
     * @param qualifier  列名
     * @param value      值
     */
    public void put(String tableName,
                    String rowName,
                    String familyName,
                    String qualifier,
                    byte[] value) {
        template.put(tableName, rowName, familyName, qualifier, value);
    }

    /**
     * 插入一行
     *
     * @param tableName  表名
     * @param rowName    rowKey
     * @param familyName 列族
     * @param qualifier  列名
     * @param value      值
     */
    public void put(String tableName,
                    String rowName,
                    String familyName,
                    String qualifier,
                    String value) {
        put(tableName, rowName, familyName, qualifier, Bytes.toBytes(value));
    }

    /**
     * 获得一行
     *
     * @param tableName 表名
     * @param rowName   rowKey
     *
     * @return
     */
    public Row get(String tableName,
                   String rowName) {
        return get(tableName, rowName, null);
    }

    /**
     * 获得一行
     *
     * @param tableName
     * @param rowName
     * @param familyName
     *
     * @return
     */
    public Row get(String tableName,
                   String rowName,
                   String familyName) {
        return get(tableName, rowName, familyName, null);
    }

    /**
     * 获得一行
     *
     * @param tableName  表名
     * @param rowName    rowKey
     * @param familyName 列族
     * @param qualifier  列名
     *
     * @return
     */
    public Row get(String tableName,
                   String rowName,
                   String familyName,
                   String qualifier) {
        return template.get(tableName, rowName, familyName, qualifier, hbaseRowMapper);
    }

    public Row get(String tableName,
                   String rowName,
                   List<String> familyNames,
                   List<String> qualifiers) {
        return get(tableName, rowName, familyNames, qualifiers, hbaseRowMapper);
    }

    public Row get(String tableName,
                   String rowName,
                   List<String> familyNames,
                   List<String> qualifiers,
                   HbaseRowMapper mapper) {
        return template.execute(tableName, new TableCallback<Row>() {
            @Override
            public Row doInTable(HTableInterface table) throws Throwable {
                Get get = new Get(rowName.getBytes(getCharset()));
                if (familyNames != null && familyNames.size() > 0) {
                    for (String family : familyNames) {
                        if (StringUtils.isNotBlank(family)) {
                            get.addFamily(Bytes.toBytes(family));
                        }
                    }
                }
                if (qualifiers != null && qualifiers.size() > 0) {
                    for (String column : qualifiers) {
                        if (StringUtils.isNotBlank(column)) {
                            String[] c = column.split(":");
                            get.addColumn(Bytes.toBytes(c[0]), Bytes.toBytes(c[1]));
                        }
                    }
                }
                Result result = table.get(get);
                return mapper.mapRow(result, 0);
            }
        });
    }

    /**
     * 统计
     *
     * @param tableName 表名
     * @param filter    过滤器
     *
     * @return
     */
    public long count(String tableName,
                      Filter filter) {
        if (filter != null) {
            FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);// FilterList.Operator.MUST_PASS_ALL -->and ，FilterList.Operator.MUST_PASS_ONE -->or
            filterList.addFilter(filter);
            return count(tableName, filterList);
        }
        return count(tableName, null);
    }

    /**
     * 统计
     *
     * @param tableName  表名
     * @param filterList 过滤器集合(并条件：FilterList.Operator.MUST_PASS_ALL或条件：FilterList.Operator.MUST_PASS_ONE)
     *
     * @return
     */
    public long count(String tableName,
                      FilterList filterList) {
        return count(tableName, filterList, null, null);
    }

    /**
     * 统计
     *
     * @param tableName
     * @param filterList
     * @param startRow
     * @param stopRow
     *
     * @return
     */
    public long count(String tableName,
                      FilterList filterList,
                      String startRow,
                      String stopRow) {
        logger.debug("startRow：" + startRow);
        logger.debug("stopRow：" + stopRow);

        Long total = 0L;
        Scan scan = getScan(1, startRow, stopRow, null, filterList, null, null);
        try {
            return countRowByCoprocessor(tableName, scan);
        } catch (Throwable e) {
            logger.error(String.format("统计%s表出错，下面尝试注册协处理器再统计!", tableName));
            e.printStackTrace();

            registerCoprocessor(tableName);// 创建协处理器

            try {
                return countRowByCoprocessor(tableName, scan);
            } catch (Throwable e1) {
                logger.error(String.format("使用协处理器，统计%s表出错!下面尝试自己循环统计总数。", tableName));
                e1.printStackTrace();

                List<Row> result;
                if (filterList == null) {
                    filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
                }
                Filter keyOnlyFilter = new KeyOnlyFilter();// 这个过滤器唯一的功能就是只返回每行的行键，值全部为空，这对于只关注于行键的应用场景来说非常合适，这样忽略掉其值就可以减少传递到客户端的数据量，能起到一定的优化作用
                filterList.addFilter(keyOnlyFilter);
                result = find(tableName, TOTALUSEPAGESIZE, true, startRow, stopRow, null, filterList, null, null);
                total += result.size();
                while (result.size() >= TOTALUSEPAGESIZE) {// 还有下一页
                    Row r = result.get(result.size() - 1);// row取上一页结果集的最后一条
                    String rowKey = r.getRowName();

                    logger.debug("startRow：" + rowKey);
                    logger.debug("stopRow：" + stopRow);

                    result = find(tableName, TOTALUSEPAGESIZE, false, rowKey, stopRow, null, filterList, null, null);
                    total += result.size();
                }
            }
        }

        return total;
    }

    /**
     * 统计数目
     *
     * @param tableName
     * @param scan
     *
     * @return
     *
     * @throws Throwable
     */
    public long countRowByCoprocessor(String tableName,
                                      Scan scan) throws Throwable {
        AggregationClient aggregationClient = new AggregationClient(template.getConfiguration());
        long total = 0;
        total = aggregationClient.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);
        aggregationClient.close();
        return total;
    }

    /**
     * 查找多行
     *
     * @param tableName 表名
     * @param pageSize  查多少条
     * @param startRow  起始rowKey
     * @param stopRow   结束rowKey
     * @param reversed  是否倒序扫描
     * @param filter    过滤器
     *
     * @return
     */
    public List<Row> find(String tableName,
                          int pageSize,
                          String startRow,
                          String stopRow,
                          Boolean reversed,
                          Filter filter) {
        if (filter != null) {
            FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);// FilterList.Operator.MUST_PASS_ALL -->and ，FilterList.Operator.MUST_PASS_ONE -->or
            filterList.addFilter(filter);
            return find(tableName, pageSize, startRow, stopRow, reversed, filterList);
        }
        return find(tableName, pageSize, startRow, stopRow, reversed, null);
    }

    /**
     * 查找多行
     *
     * @param tableName  表名
     * @param pageSize   查多少条
     * @param startRow   起始rowKey
     * @param stopRow    结束rowKey
     * @param reversed   是否倒序扫描
     * @param filterList 过滤器集合(并条件：FilterList.Operator.MUST_PASS_ALL或条件：FilterList.Operator.MUST_PASS_ONE)
     *
     * @return
     */
    public List<Row> find(String tableName,
                          int pageSize,
                          String startRow,
                          String stopRow,
                          Boolean reversed,
                          FilterList filterList) {
        return find(tableName, pageSize, true, startRow, stopRow, reversed, filterList, null, null);
    }

    /**
     * 查找多行
     *
     * @param tableName   表名
     * @param pageSize    每页查多少条
     * @param isFirstPage 是否查第一页
     * @param startRow    起始rowKey
     * @param stopRow     结束rowKey
     * @param reversed    是否倒序扫描
     * @param filterList  过滤器集合
     * @param familyNames 要查询的列族集合
     * @param qualifiers  要查询的列集合，格式(family:column)
     *
     * @return
     */
    public List<Row> find(String tableName,
                          Integer pageSize,
                          Boolean isFirstPage,
                          String startRow,
                          String stopRow,
                          Boolean reversed,
                          FilterList filterList,
                          List<String> familyNames,
                          List<String> qualifiers) {
        return template.execute(tableName, new TableCallback<List<Row>>() {
            @Override
            public List<Row> doInTable(HTableInterface table) throws Throwable {
                boolean isKeyOnlyFilter = false;// 是否是只返回rowKey的filter
                if (filterList != null) {
                    List<Filter> filters = filterList.getFilters();
                    for (Filter filter : filters) {
                        if (filter.toString().equals("KeyOnlyFilter")) {
                            isKeyOnlyFilter = true;
                            break;
                        }
                    }
                }

                List<Row> result = new ArrayList<>();
                Boolean firstPage = true;// 默认查第一页
                if (isFirstPage != null) {
                    firstPage = isFirstPage;
                }

                Scan scan = getScan(pageSize, startRow, stopRow, reversed, filterList, familyNames, qualifiers);

                ResultScanner scanner = table.getScanner(scan);
                try {
                    Result[] rs;
                    if (firstPage) {
                        rs = scanner.next(pageSize);
                    } else {
                        rs = scanner.next(pageSize + 1);// 如果不是第一页，那么多查出一条
                    }
                    for (Result r : rs) {
                        result.add(HbaseRowMapper.resultToRow(r, isKeyOnlyFilter));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    scanner.close();
                }
                if (!firstPage && result.size() > 0) {// 由于是第N+1页，上面多查了一条记录，所以这里删掉多的那一条
                    result.remove(0);// 将第一条删除
                }
                return result;
            }

        });
    }

    /**
     * 删除一行
     *
     * @param tableName
     * @param rowName
     * @param familyName
     * @param qualifier
     */
    public void delete(String tableName,
                       String rowName,
                       String familyName,
                       String qualifier) {
        template.delete(tableName, rowName, familyName, qualifier);
    }

    /**
     * 获得扫描类
     *
     * @param pageSize    每页显示记录数
     * @param startRow    起始rowKey
     * @param stopRow     终止rowKey
     * @param reversed    是否倒序扫描
     * @param filterList  过滤器集合
     * @param familyNames 要查询的列族集合
     * @param qualifiers  要查询的列集合，格式(family:column)
     *
     * @return
     */
    private Scan getScan(Integer pageSize,
                         String startRow,
                         String stopRow,
                         Boolean reversed,
                         FilterList filterList,
                         List<String> familyNames,
                         List<String> qualifiers) {
        Scan scan = new Scan();// 创建扫描对象

        if (pageSize > 1) {
            scan.setCaching(pageSize + 1);// 设置的值为每次rpc的请求记录数，默认是1；cache大可以优化性能，但是太大了会花费很长的时间进行一次传输
        }
        scan.setCacheBlocks(true);// Scan实例可以在RegionServer中使用块缓存，可以由setCacheBlocks方法控制。如果Scan是MapReduce的输入源，要将这个值设置为 false。对于经常读到的行，就建议使用块缓冲。
        // scan.setBatch(10000);//设置每次取的column size；有些row特别大，所以需要分开传给client，就是一次传一个row的几个column。
        // scan.setMaxResultSize(10000);

        if (reversed != null && reversed) {// 倒序扫描
            scan.setReversed(reversed);
        }
        if (StringUtils.isNotBlank(startRow)) {
            scan.setStartRow(Bytes.toBytes(startRow));
        }
        if (StringUtils.isNotBlank(stopRow)) {
            scan.setStopRow(Bytes.toBytes(stopRow));
        }
        if (filterList != null) {
            scan.setFilter(filterList);
        }
        if (familyNames != null && familyNames.size() > 0) {
            for (String family : familyNames) {
                if (StringUtils.isNotBlank(family)) {
                    scan.addFamily(Bytes.toBytes(family));
                }
            }
        }
        if (qualifiers != null && qualifiers.size() > 0) {
            for (String column : qualifiers) {
                if (StringUtils.isNotBlank(column)) {
                    String[] c = column.split(":");
                    scan.addColumn(Bytes.toBytes(c[0]), Bytes.toBytes(c[1]));
                }
            }
        }
        return scan;
    }

}
